package com.sailing.lianxi.demoservice.onecusmer;

import java.io.BufferedWriter;
import java.io.FileWriter;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.stream.Collectors;
import java.util.stream.Stream;

public class OffsetManager {

	private String storagePrefix;
	
	public OffsetManager(String storagePrefix) {
		this.storagePrefix = storagePrefix;
	}
	
	public void saveOffsetInExternalStore(String topic,int partition,long offset) {
		try {
			 FileWriter fileWriter = new FileWriter(storageName(topic, partition));
			 BufferedWriter bufferedWriter = new BufferedWriter(fileWriter);
			 bufferedWriter.write(offset + "");
			 bufferedWriter.flush();
			 bufferedWriter.close();
		} catch (Exception e) {
			e.printStackTrace();
			throw new RuntimeException(e);
		}
	}
	
	public long readOffsetFromExternalStore(String topic, int partition) {
		try {
			Stream<String> lines = Files.lines(Paths.get(storageName(topic,partition)));
			return Long.parseLong((String) lines.collect(Collectors.toList()).get(0)) + 1;
		} catch (IOException e) {
			e.printStackTrace();
		}
		return 0;
	}
	
	 private String storageName(String topic, int partition) {
         return storagePrefix + "-" + topic + "-" + partition;
     }
}
